package com.ecarx.camera.utils.rx;

import com.ecarx.camera.utils.AppExecutors;

import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;

/**
 * RxJava相关工具类
 */
public class RxUtils {

    private static final long DEFAULT_TIMEOUT = 5;
    private static final TimeUnit DEFAULT_TIME_UNIT = TimeUnit.SECONDS;

    /**
     * 将一般操作转换成返回Disposable的异步请求
     * @param action 耗时操作
     * @param parameters 耗时操作的参数
     * @param <T> 返回值类型
     * @return
     */
    public static <T> Disposable justRun(ObservableAction<T> action, Object... parameters) {
        return doAction(action, null, parameters);
    }

    /**
     * 将一般操作转换成返回Disposable的异步请求
     * @param action 耗时操作
     * @param callback UI回调
     * @param parameters 耗时操作的参数
     * @param <T> 返回值类型
     * @return
     */
    public static <T> Disposable doAction(ObservableAction<T> action, ObservableCallback<T> callback, Object... parameters) {
        return doActionWithTimeout(action, callback, DEFAULT_TIMEOUT, DEFAULT_TIME_UNIT, parameters);
    }

    /**
     * 将一般操作转换成返回Disposable的异步请求
     * @param action 耗时操作
     * @param callback UI回调
     * @param timeout 超时时间
     * @param timeUnit 超时单位
     * @param parameters 耗时操作的参数
     * @param <T> 返回值类型
     * @return
     */
    public static <T> Disposable doActionWithTimeout(ObservableAction<T> action, ObservableCallback<T> callback, long timeout, TimeUnit timeUnit, Object... parameters) {
        return Observable.create(new ObservableOnSubscribe<Action>() {

                    @Override
                    public void subscribe(ObservableEmitter<Action> emitter) throws Exception {
                        if (action != null) {
                            final T data = action.run(parameters);
                            emitter.onNext(new Action() {
                                @Override
                                public void run() throws Exception {
                                    if (callback != null) {
                                        callback.onCallback(data);
                                    }
                                }
                            });
                            emitter.onComplete();
                        }
                    }
                }).compose(RxTransformer.switchSchedulers())
                .timeout(timeout, timeUnit)
                .subscribe(new Consumer<Action>() {
                    @Override
                    public void accept(Action action) throws Exception {
                        action.run();
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        AppExecutors.get().mainThread().execute(new Runnable() {
                            @Override
                            public void run() {
                                if (callback != null) {
                                    callback.onThrowable(throwable);
                                }
                            }
                        });
                    }
                });
    }

    /**
     * 耗时的操作接口定义
     * @param <T> 返回值
     */
    public interface ObservableAction<T> {
        /**
         * 耗时方法
         * @param parameters 参数
         * @return 返回值
         */
        T run(Object... parameters);
    }

    /**
     * UI回调接口定义
     * @param <T>
     */
    public interface ObservableCallback<T> {
        /**
         * 数据回调
         * @param result
         */
        void onCallback(T result);

        /**
         * 异常回调
         * @param throwable
         */
        void onThrowable(Throwable throwable);
    }
}
